package org.orqlog.client;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.data.Stat;
import org.orqlog.encoder.RpcRequest;
import org.orqlog.service.UserService;

import java.lang.reflect.Array;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;

public class ClientBootstrap {
    public static  final String providerName="UserService#sayHello#";
    private CuratorFramework client;
    private String path = "/lg-server";
    private HashMap<String, UserService> consumers = new HashMap<>();
    private Date makingConsumers = null;
    List<String> tServers = new ArrayList<>();

    public ClientBootstrap() {
        client = CuratorFrameworkFactory.builder()
                .connectString("127.0.0.1:2181")
                .sessionTimeoutMs(5000)
                .connectionTimeoutMs(3000)
                .retryPolicy(new ExponentialBackoffRetry(1000, 5))
                .build();

        client.start();
        System.out.println("Zookeeper session established.");
    }

    public List<String> getAliveServer() throws Exception {
        String path = "/lg-server";
        List<String> servers = client.getChildren().forPath(path);
        return servers;
    }


    /**
     * 监听zk服务器注册节点
     * @throws Exception
     */
    public void watchNode() throws Exception {
        PathChildrenCache pathChildrenCache = new PathChildrenCache(client, path, true);
        PathChildrenCacheListener pathChildrenCacheListener = new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {
                ChildData childData = pathChildrenCacheEvent.getData();
                switch (pathChildrenCacheEvent.getType()) {
                    case CHILD_ADDED:
                        System.out.println(new Date() +  "子节点增加，path=" + childData.getPath() + ",data=" + new String(childData.getData(), "utf-8"));
                        makeConsumers();
                        break;
                    case CHILD_UPDATED:
                        System.out.println("子节点更新，path=" + childData.getPath() + ",data=" + new String(childData.getData(), "utf-8"));
                        break;
                    case CHILD_REMOVED:
                        System.out.println(new Date() + "子节点删除，path=" + childData.getPath() + ",data=" + new String(childData.getData(), "utf-8"));
                        makeConsumers();
                        break;
                    default:
                        break;
                }
            }
        };

        pathChildrenCache.getListenable().addListener(pathChildrenCacheListener);
        pathChildrenCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
    }

    protected String getResponseTime(String addr) throws Exception {
        byte[] bytes = client.getData().forPath("/lg-server-restime/"+addr);
        return new String(bytes);
    }

    /**
     * 创建consumer的proxy
     * @return
     * @throws Exception
     */
    public void makeConsumers() throws Exception {

        makingConsumers = new Date();
        System.out.println(">>>Making consumers...........");
        List<String> aServers = getAliveServer();
        HashMap<String, UserService> aConsumers = new HashMap<>();

        for (String addr : aServers) {
            RpcConsumer rpcConsumer = new RpcConsumer(addr);
            UserService proxy = (UserService) rpcConsumer.createProxy(UserService.class);
            aConsumers.put(addr, proxy);
        }
        consumers = aConsumers;

        this.tServers = aServers;
        System.out.println(">>>>Current consumer count: " + consumers.size());
    }

    public HashMap<String, UserService> getConsumers() {
        return consumers;
    }

    public List<String> getTservers() {
        return tServers;
    }

    public String selectConsumer(List<String> servers, HashMap<String, UserService> consumers) throws Exception {
        String tAddr = null;
        Float pTime = Float.POSITIVE_INFINITY;
        Float bTime = 0f;
        String zTime[];
        long now = (new Date()).getTime();
        for (String addr : servers) {
            // 查找反应时间最短的服务端进行连接
            System.out.println(">>> response time: " + getResponseTime(addr));
            zTime = getResponseTime(addr).split(",");
            if (now - (Long.parseLong(zTime[0])) > 5000L) {
                tAddr = addr;
                break;
            }

            bTime = Float.parseFloat(zTime[1]);
            if (bTime < pTime) {
                tAddr = addr;
            }
        }

        return tAddr;
    }

    public static void main(String[] args) throws Exception {
        ClientBootstrap cBootstrap = new ClientBootstrap();
        cBootstrap.makeConsumers();
        HashMap<String, UserService> consumers;
        RpcRequest rpcRequest = null;
        cBootstrap.watchNode();
        String tAddr = null;

        while (true) {
            // 监听zk服务器注册节点
            Thread.sleep(2000);
//            System.out.println("_____________" );
            consumers = cBootstrap.getConsumers();

            // 发送的请求对象
            rpcRequest = new RpcRequest();
            rpcRequest.setClassName("org.orqlog.service.UserServiceImpl");
            rpcRequest.setMethodName("methodToCall");
            String[] tArgs = {"The Args"};
            rpcRequest.setParameters(tArgs);

            // 发送请求
            List<String> tServers = cBootstrap.getTservers();
            // 第一题
            for (String addr : tServers) {
                System.out.println(consumers.get(addr).sayHello(rpcRequest));
            }

            // 第二题
//            tAddr = cBootstrap.selectConsumer(tServers, consumers);
//            if (tAddr != null) {
//                System.out.println(consumers.get(tAddr).sayHello(rpcRequest));
//            }



        }

    }
}
